// 文件路径: dbms/execution/execution/ScatterGatherExecutor.java
package com.dbms.execution.execution;

import com.dbms.catalog.Catalog;
import com.dbms.cluster.ClusterManager;
import com.dbms.cluster.ClusterRequest;
import com.dbms.cluster.NodeInfo;
import com.dbms.cluster.PeerNodeClient;
import com.dbms.common.QueryResult;
import com.dbms.common.Schema;
import com.dbms.common.Tuple;
import com.dbms.execution.plan.PlanNode;
import com.dbms.parser.ast.SelectNode;

import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.stream.Collectors;

/**
 * 【已修复】
 * ScatterGatherExecutor (分散-聚集执行器) 是分布式查询的协调器。
 * 它的工作方式是：
 * 1. 分散 (Scatter): 将一个查询请求并行地发送到集群中的所有工作节点。
 * 2. 聚集 (Gather): 异步地等待并收集来自所有工作节点的查询结果，然后将它们合并成一个单一的结果集。
 *
 * 核心修复：这个版本现在可以在查询规划阶段（构造时）就从协调器节点的本地 Catalog 中确定查询的 Schema，
 * 从而避免了在执行期间因无法确定 Schema 而导致的 NullPointerException。
 */
public class ScatterGatherExecutor implements Executor {

    private final SelectNode selectNode; // 查询的AST节点
    private final String sql;            // 发送到工作节点的SQL语句
    private final String currentUser;
    private final Catalog catalog;       // 【新增】协调器节点的目录管理器
    private final Schema schema;         // 【新增】在构造时就确定的查询结果模式
    private Iterator<Tuple> resultIterator; // 最终结果集的迭代器
    private List<Tuple> gatheredTuples; // 从所有节点收集到的元组列表

    public ScatterGatherExecutor(SelectNode selectNode, String currentUser, Catalog catalog) {
        this.selectNode = selectNode;
        this.currentUser = currentUser;
        this.catalog = catalog; // 【新增】
        // 【核心修正】在构造函数中，立即从协调器本地的 Catalog 获取表的 Schema。
        // 这确保了即使在查询执行之前，Schema 也总是可用的。
        this.schema = catalog.getTable(selectNode.getTableName()).schema;
        // 【优化】更精确地从 AST 重建 SQL，以发送给工作节点
        this.sql = buildSqlFromNode(selectNode);
    }

    /**
     * 【优化】辅助方法，用于从 AST 节点重建一个简单的 SELECT SQL 语句。
     * @param node SELECT 查询的 AST 节点
     * @return 重建的 SQL 字符串
     */
    private String buildSqlFromNode(SelectNode node) {
        StringBuilder sqlBuilder = new StringBuilder("SELECT ");
        String columns = node.getColumns().stream()
                .map(Object::toString)
                .collect(Collectors.joining(", "));
        sqlBuilder.append(columns);
        sqlBuilder.append(" FROM ");
        sqlBuilder.append(node.getTableName());
        // 注意：此执行器当前仅用于全表扫描，因此不重建 WHERE 等更复杂的子句
        sqlBuilder.append(";");
        return sqlBuilder.toString();
    }

    @Override
    public void open() {
        System.out.println("Scatter-Gather: Starting execution for query: " + sql);
        this.gatheredTuples = new ArrayList<>();

        ClusterManager cm = ClusterManager.getInstance();
        List<NodeInfo> allNodes = new ArrayList<>(cm.getClusterNodes().values());

        // 创建一个固定大小的线程池，为每个节点分配一个线程
        ExecutorService threadPool = Executors.newFixedThreadPool(allNodes.size());
        List<Future<QueryResult>> futures = new ArrayList<>();

        ClusterRequest request = new ClusterRequest(sql, currentUser);

        // --- 分散 (Scatter) 阶段 ---
        for (NodeInfo node : allNodes) {
            Future<QueryResult> future = threadPool.submit(() -> {
                System.out.println("Scatter-Gather: Sending query to node " + node.getNodeId());
                try {
                    // 使用 PeerNodeClient 向远程节点发送请求
                    return PeerNodeClient.sendRequest(node, request).getQueryResult();
                } catch (Exception e) {
                    System.err.println("Scatter-Gather: Failed to get result from node " + node.getNodeId() + ": " + e.getMessage());
                    // 如果请求失败，返回一个包含错误信息的 QueryResult
                    return new QueryResult("Error from node " + node.getNodeId() + ": " + e.getMessage());
                }
            });
            futures.add(future);
        }

        // --- 聚集 (Gather) 阶段 ---
        // 遍历 futures 列表，阻塞等待每个节点的响应
        for (Future<QueryResult> future : futures) {
            try {
                QueryResult result = future.get(); // get() 方法会阻塞直到任务完成
                if (result != null && result.isSuccess) {
                    // 如果节点成功返回结果，将其元组添加到总列表中
                    gatheredTuples.addAll(result.tuples);
                } else if (result != null) {
                    System.err.println("Scatter-Gather: Received error response: " + result.message);
                }
            } catch (Exception e) {
                System.err.println("Scatter-Gather: Exception while gathering results: " + e.getMessage());
            }
        }

        threadPool.shutdown(); // 关闭线程池
        this.resultIterator = gatheredTuples.iterator(); // 初始化结果迭代器
        System.out.println("Scatter-Gather: Finished. Gathered " + gatheredTuples.size() + " tuples from all nodes.");
    }

    @Override
    public Tuple next() {
        if (resultIterator != null && resultIterator.hasNext()) {
            return resultIterator.next();
        }
        return null;
    }

    @Override
    public void close() {
        this.resultIterator = null;
        this.gatheredTuples = null;
    }

    @Override
    public Schema getSchema() {
        // 【核心修正】直接返回在构造时就已经确定的 Schema
        return this.schema;
    }

    @Override
    public PlanNode getPlanNode() {
        return null;
    }
}